package com.intct.func;

import com.intct.hbase.bean.LocusRedisBean;
import com.intct.util.RedisUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

/**
 * @author gufg
 * @since 2025-07-28 16:37
 */
public class RedisSinkFunction extends RichSinkFunction<LocusRedisBean> {

    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("开始连接Redis......");
        jedis = RedisUtil.getConnection();
        System.out.println("连接Redis完成......");
    }

    @Override
    public void invoke(LocusRedisBean locusRedisBean, Context context) throws Exception {
        System.out.println("开始保存数据到Redis......");
        String key = locusRedisBean.getOrderId();
        long score = locusRedisBean.getScore();
        String value = locusRedisBean.getValue();

        // 如果当前订单结束
        if ("end:end".equals(value)) {
            // 在当前订单集合删除掉
            jedis.srem("realtime", key);
            // 历史订单集合
            jedis.sadd("history", key);
        } else {
            // 当前订单集合
            jedis.sadd("realtime", key);
        }
        // 订单轨迹
        jedis.zadd(key, score, value);
        System.out.println("结束保存数据到Redis......");
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeConnection(jedis);
    }
}
